#include <linux/init.h>
#include <linux/module.h>
#include <linux/types.h>
#include <linux/blkdev.h>
#include <linux/fs.h>
#include <linux/genhd.h>
#include <linux/bio.h>
#include <linux/hdreg.h>
#include <linux/net.h>
#include <linux/kthread.h>
#include <linux/semaphore.h>
#include <linux/mutex.h>
#include <linux/sched.h>
#include <linux/list.h>
#include <linux/in.h>
#include <net/sock.h>
#include <linux/completion.h>
#include <linux/kdev_t.h>
#include "kblock_proc.h"
#include "kblock_debugfs.h"
#include "kblock.h"
#include "kblock_receive.h"
#include "kblock_send.h"


MODULE_LICENSE("GPL");
MODULE_AUTHOR("duxing");
MODULE_DESCRIPTION("distributed block deviver");

static int kblock_make_request(struct request_queue *queue,struct bio *bio);
static int kblock_open(struct block_device *device,fmode_t mode);
static int kblock_release(struct gendisk *disk,fmode_t mode);
static int kblock_getgeo(struct block_device *device,struct hd_geometry *geo);
static int kblock_local_bio(struct kblock_device *device,struct bio *bio);

struct kblock_device *kdevice = NULL;
int role = KBLOCK_MASTER;
int debug = 0;
char *physics_disk = "/dev/sdb1";


struct block_device_operations kblock_fops = {
    .open = kblock_open,
    .release = kblock_release,
    .getgeo = kblock_getgeo,
    .owner = THIS_MODULE,
};

void kblock_bio_endio(struct bio *bio,int error)
{
    //struct kblock_device *device = bio->bi_private;
    int uptodate = bio_flagged(bio,BIO_UPTODATE);
    int is_write = bio_data_dir(bio) == WRITE;

    if (!error && !uptodate) {
        printk(KERN_ERR "%s:bio is error or not uptodate\n",__FUNCTION__);
        error = -EIO;
    }

    bio_put(bio);
    if (is_write) {
        printk(KERN_INFO "%s:bio is writed\n",__FUNCTION__);
    }else {
        printk(KERN_INFO "%s:not support READ now\n",__FUNCTION__);
    }
    return;
}

static int kblock_local_bio(struct kblock_device *device,struct bio *bio)
{
    if (device->flag == KBLOCK_SLAVE && bio->bi_rw != WRITE) {
		printk(KERN_INFO "slave not handle read operation\n");
		return 0;
    }
	bio->bi_bdev = device->physics_device;
	generic_make_request(bio);
	return 0;
}
static int kblock_make_request(struct request_queue *queue,struct bio *bio)
{
    struct kblock_device *device = NULL;

    device = queue->queuedata;

    if(atomic_read(&device->status) != KBLOCK_CONNECTED) {
        printk(KERN_ERR "kblock not connected\n");
		return -1;
    }
    switch(device->flag) {
        case KBLOCK_MASTER:
            kblock_remote_bio(device,bio);
            kblock_local_bio(device,bio);
            break;
        case KBLOCK_SLAVE:
            kblock_local_bio(device,bio);
            break;
        default:
            printk(KERN_ERR "bad role of device\n");
            return -1;
    }
    return 0;
}

static int kblock_open(struct block_device *device,fmode_t mode)
{
    return 0;
}

static int kblock_release(struct gendisk *disk,fmode_t mode)
{
    return 0;
}

static int kblock_getgeo(struct block_device *device,struct hd_geometry *geo)
{
    geo->heads = 1;
    geo->sectors = 1;
    geo->cylinders = (KBLOCK_CAPACITY>>9) /geo->heads/geo->sectors;
    return 0;
}

static int kblock_try_connect(struct kblock_device *device)
{
    struct socket *sock = NULL;
    struct sockaddr_in remote;
    int err = 0;

    memset(&remote,0,sizeof(remote));
    remote.sin_family = AF_INET;
    remote.sin_port = htons(KBLOCK_PORT);
    remote.sin_addr.s_addr = in_aton(KBLOCK_ADDR);

    if (sock_create(AF_INET,SOCK_STREAM,0,&sock) < 0) {
        printk(KERN_ERR "socket_create failed\n");
        return -1;
    }
    
    sock->sk->sk_rcvtimeo = 1 * HZ;
    sock->sk->sk_sndtimeo = 1 * HZ;
    if ((err = sock->ops->connect(sock,(struct sockaddr*)&remote,sizeof(remote),0)) < 0) {
        printk(KERN_ERR "socket_connect failed\n");
        if (sock != NULL) {
            sock_release(sock);
        }
        switch(-err) {
            #ifdef ETIMEOUT
            case ETIMEOUT:
            #endif
            case EAGAIN:
            case EINTR:
            case EINPROGRESS:
            case ERESTARTSYS:
            case ECONNREFUSED:
            case EHOSTDOWN:
            case EHOSTUNREACH:
                printk(KERN_ERR "connect failed errno %d",-err);
                break;
            default:
                break;
        }
        return -1;
    }

    device->data_sock = sock;
    return 0;
}

static int kblock_try_wait_connect(struct kblock_device *device)
{
    struct socket *sock = NULL;
    struct socket *newsk;
    struct sockaddr_in local;
    int err = 0;

    memset(&local,0,sizeof(local));
    local.sin_family = AF_INET;
    local.sin_port = htons(KBLOCK_PORT);
    local.sin_addr.s_addr = in_aton(KBLOCK_ADDR);

    if (sock_create(AF_INET,SOCK_STREAM,0,&sock) < 0) {
        printk(KERN_ERR "socket_create failed\n");
        return -1;
    }
    sock->sk->sk_reuse = 1;
    sock->sk->sk_rcvtimeo = 1 * HZ;
    sock->sk->sk_sndtimeo = 1 * HZ;
    device->listen_sock = sock;

    if (sock->ops->bind(sock,(struct sockaddr*)&local,sizeof(local)) < 0) {
        printk(KERN_ERR "socket bind failed\n");
        goto failed;
    }

    if (sock->ops->listen(sock,5) < 0) {
        printk(KERN_ERR "listen failed\n");
        goto failed;
    }
//again:
    if ((err = kernel_accept(sock,&newsk,0)) < 0 ) {
        printk(KERN_ERR "accept failed.errno %d\n",err);
        goto failed;
    }
    printk(KERN_INFO "accept success\n");
    newsk->ops = sock->ops;
    device->data_sock = newsk;
    return 0;
failed:
    if (sock != NULL) {
        sock_release(sock);
        device->listen_sock = NULL;
    }
    return -1;
}

static int kthread_work(void *arg)
{
    int error = 0;
    struct kblock_device *device = NULL;
    device = (struct kblock_device*)arg;

    printk(KERN_INFO "thread kblock_work staring\n");

again:    
    if (atomic_read(&device->status) == KBLOCK_EXIT) {
        printk(KERN_INFO "kblock_work thread exit\n");
        complete(&device->work_completion);
        return 0;
    }
    if (device->flag == KBLOCK_MASTER) {
        error = kblock_try_wait_connect(device);
        if (error == 0) {
            printk(KERN_INFO "wait connect sucess\n");
            goto out;
        }
        schedule_timeout_interruptible(2 * HZ);
        goto again;
    }
    if (device->flag == KBLOCK_SLAVE) {
        error = kblock_try_connect(device);
        if (error == 0) {
            printk(KERN_INFO "connect success\n");
            goto out2;
        }
        schedule_timeout_interruptible(2 * HZ);
        goto again;
    }
out:
	atomic_set(&kdevice->status,KBLOCK_CONNECTED);
    device->thread_work = NULL;
    return 0;
out2:
	atomic_set(&kdevice->status,KBLOCK_CONNECTED);
	do {
		if (atomic_read(&device->status) == KBLOCK_EXIT) {
            printk(KERN_INFO "kblock_work thread exit\n");
            complete(&device->work_completion);
            return 0;
        }
		if (kblock_recv_bio(device) < 0) {
			printk(KERN_INFO "kblock_recv_bio failed\n");
			break;
		}
	}while(1);
	return 0;
}

static int __init kblock_init(void)
{
    struct task_struct *nt;
    sector_t size = 0;

    kdevice = kmalloc(sizeof(struct kblock_device),GFP_KERNEL);
    if (kdevice == NULL) {
        printk(KERN_ERR "%s:malloc kblock_device failed\n",__FUNCTION__);
        return -1;
    }
    memset(kdevice,0,sizeof(struct kblock_device));

    kdevice->flag = role;

    if ((kdevice->major = register_blkdev(0,KBLOCK_DEVNAME)) < 0) {
        goto failed;
    }

    kdevice->physics_device = blkdev_get_by_path(physics_disk, FMODE_READ|FMODE_WRITE,NULL);

    if (kdevice->physics_device == NULL) {
        printk(KERN_INFO "lookup_bdev %s failed\n",BACK_DISK);
        goto failed;
    }
    if ((kdevice->queue = blk_alloc_queue(GFP_KERNEL)) == NULL) {
        goto failed;
    }
    blk_queue_make_request(kdevice->queue,kblock_make_request);
    kdevice->queue->queuedata = kdevice;

    if ((kdevice->vdisk = alloc_disk(1)) == NULL) {
        goto failed;
    }
    strcpy(kdevice->vdisk->disk_name,KBLOCK_DEVNAME);
    kdevice->vdisk->major = kdevice->major;
    kdevice->vdisk->first_minor = 0;
    kdevice->vdisk->fops = &kblock_fops;
    kdevice->vdisk->queue = kdevice->queue;
    kdevice->vdisk->private_data = kdevice;

    size = get_capacity(kdevice->physics_device->bd_disk);
    printk(KERN_INFO "kblock:%s sector_t %ld",physics_disk,(long)size);
    set_capacity(kdevice->vdisk,size);
    add_disk(kdevice->vdisk); 

    init_completion(&kdevice->work_completion);

    nt = kthread_create(kthread_work,kdevice,"kblock_work");
    if (IS_ERR(nt)) {
        goto failed;
    }
    kdevice->thread_work = nt;
    wake_up_process(kdevice->thread_work);
	
    if (register_kblock_proc() < 0) {
		goto failed;
    }

	if (register_kblock_debugfs() < 0) {
		goto failed;
	}
    return 0;
failed:
    if (kdevice->major !=0 ) {
        unregister_blkdev(kdevice->major,KBLOCK_DEVNAME);
    }

    if (kdevice->vdisk != NULL) {
        del_gendisk(kdevice->vdisk);
    }

    if (kdevice->queue != NULL) {
        blk_cleanup_queue(kdevice->queue);
    }

    kfree(kdevice);
    return -1;
}

static void __exit kblock_exit(void)
{
    atomic_set(&kdevice->status,KBLOCK_EXIT);

    if (kdevice->thread_work != NULL) {
        wait_for_completion(&kdevice->work_completion);
    }

    if (kdevice->data_sock != NULL) {
        sock_release(kdevice->data_sock);
    }

    if (kdevice->listen_sock != NULL) {
        sock_release(kdevice->listen_sock);
    }

    unregister_blkdev(kdevice->major,KBLOCK_DEVNAME); 
    del_gendisk(kdevice->vdisk);
    blk_cleanup_queue(kdevice->queue);
    kfree(kdevice);
    unregister_kblock_proc();
	unregister_kblock_debugfs();
    return;
}


module_init(kblock_init);
module_exit(kblock_exit);
module_param(role,int,S_IRUGO);
module_param(debug,int,S_IRUGO);
module_param(physics_disk,charp,S_IRUGO);
